package com.person.server;

import com.person.server.grpcjava.*;
import io.grpc.Channel;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.time.LocalDate;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


/**
 * 详细介绍，grpc这四种请求方式
 * 1.客户端非流式，服务端非流式(客户端发送一次消息，服务端接收一次消息。服务端可响应一条消息，客户端接收一次响应消息）
 * 2.客户端非流式，服务端流式请求(客户端发送一次，服务端接收一次消息。 服务端可响应多条消息onNext，当服务端onCompleted时，客户端在迭代器一次收到多条响应消息）
 * 3.客户端流式，服务端非流式(客户端发送多次onNext，服务器接收多次消息。服务端可响应一条消息，客户端接收一次响应消息）
 * 4.客户端流式，服务端流式(客户端发送多次onNext，服务器接收多次消息onNext。客户端和服务端都无需onCompleted，服务端响应多条消息onNext，客户端接收多次响应消息）
 * <p>
 * 说明：流式即消息多次发送。非流式只能发送一次
 */
public class GrpcClient {
    public static void main(String[] args) {

        GrpcClient grpcClient = new GrpcClient();
//        System.out.println("客户端和服务端都非流式请求");
//        grpcClient.getRealNameByUsername();
//
//        System.out.println("服务端流式实现");
//        grpcClient.getStudentsByAge();
//
//        System.out.println("客户端流实现");
//        grpcClient.getStudentsWrapperByAges();

    }

    /**
     * 客户端和服务端都非流式请求
     * 客户端非流式都可以用PersonServiceGrpc.newBlockingStub(managedChannel).getMethod获取到Response对象，都可以直接使用Request.newBuilder().set设置对象消息
     * 服务端和客户端都只能发一次
     */
    public void getRealNameByUsername() {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 8899)
                .usePlaintext().build();
        PersonServiceGrpc.PersonServiceBlockingStub blockingStub = PersonServiceGrpc.newBlockingStub(managedChannel);
        MyResponse myResponse = blockingStub.getRealNameByUsername(MyRequest.newBuilder().setUsername("公众号:霸道的程序猿").build());
        System.out.println(myResponse.getRealname());
    }

    /***
     * 客户端非流式，服务端流式实现
     * 客户端非流式都可以用PersonServiceGrpc.newBlockingStub(managedChannel).getMethod获取到Response对象，都可以直接使用Request.newBuilder().set设置对象消息
     * 服务端可以发多次，服务端onCompleted后，客户端收到后从迭代器遍历值
     */
    public void getStudentsByAge() {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 8899)
                .usePlaintext().build();
        PersonServiceGrpc.PersonServiceBlockingStub blockingStub = PersonServiceGrpc.newBlockingStub(managedChannel);

        System.out.println("请求-流式响应，调用getRealNameByUsername");
        Iterator<StudentResponse> iter = blockingStub.getStudentsByAge(StudentRequest.newBuilder().setAge(20).build());
        while (iter.hasNext()) {
            StudentResponse studentResponse = iter.next();
            System.out.println(studentResponse.getName());
            System.out.println(studentResponse.getAge());
            System.out.println(studentResponse.getCity());
        }
    }

    /**
     * 客户端流式，服务端非流式
     * 客户端必须使用异步的stub
     * 客户端必须实现StreamObserver接口，用于获取响应消息内容
     * 客户端发送多次，服务器也获取多次消息。服务器响应一次消息内容
     */
    public void getStudentsWrapperByAges() {
        //只有当流结束或者发生异常时才终止,不然就一直等待，可以在调用时判断时间防止一直等待
        final CountDownLatch finishLatch = new CountDownLatch(1);

        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 8899)
                .usePlaintext().build();

        PersonServiceGrpc.PersonServiceStub stub = PersonServiceGrpc.newStub(managedChannel);

        System.out.println("-----------------------------");
        System.out.println("流式请求-响应，调用GetStudentsWrapperByAges");

        //定义并实现StreamObserver接口streamObserverImpl为接口实现类
        StreamObserver<StudentResponseList> streamObserverImpl = new StreamObserver<StudentResponseList>() {
            //消息内容
            public void onNext(StudentResponseList studentResponseList) {
                studentResponseList.getStudentResponseList().forEach(studengResponse -> {
                    System.out.println(studengResponse.getName());
                    System.out.println(studengResponse.getAge());
                    System.out.println(studengResponse.getCity());
                });
            }

            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }

            public void onCompleted() {
                System.out.println("completed");
            }
        };

        //实现后的streamObserverImpl类，用来做入参。
        StreamObserver<StudentRequest> studentRequestStreamObserver = stub.getStudentsWrapperByAges(streamObserverImpl);
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(20).build());
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(30).build());
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(40).build());
        System.out.println("消息发送已发送");
        //调用onCompleted方法，通知服务端我已发完，请处理响应消息。
        studentRequestStreamObserver.onCompleted();
        System.out.println("消息发送已onCompleted，等待服务器返回");

        /**
         * 阻塞直到结束，建议加上超时时间 eg : finishLatch.await(1,TimeUnit.SECONDS)
         */
        try {
            System.out.println("阻塞直到流返回");
            finishLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("方法结束");
    }



    /**
     * 双向流式实现
     * 客户端流式，服务端流式
     * 客户端必须使用异步的stub
     * 客户端必须实现StreamObserver接口，用于获取响应消息内容
     * 客户端和服务端均无需onCompleted，即可发送消息
     */
    //重连次数
    int RETRY_COUNT = 5;
    public void biTalk() {
        //重连的定时任务
        ScheduledExecutorService timer = Executors.newScheduledThreadPool(1);

        //等待时需要阻塞
        final boolean[] firstConnectedFlag = {false};
        final CountDownLatch firstConnectedLatch = new CountDownLatch(1);

        //连接
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 8899)
                .usePlaintext().build();
        PersonServiceGrpc.PersonServiceStub stub = PersonServiceGrpc.newStub(managedChannel);

        System.out.println("-----------------------------");
        System.out.println("流式请求-流式响应，调用BiTalk");

        //定义并实现StreamObserver接口streamObserverImpl为接口实现类，如果连接失败次处的streamObserverImpl不处理消息
        StreamObserver<StreamResponse> streamObserverImpl = new StreamObserver<StreamResponse>() {
            @Override
            public void onNext(StreamResponse streamResponse) {
                // 启动成功，连接已建立（之后可以把服务器断开，客户端需要一直重试，直到服务器重启成功）
                System.out.println(streamResponse.getResponseInfo());
                firstConnectedFlag[0] = true;
                firstConnectedLatch.countDown();
                RETRY_COUNT = Integer.MAX_VALUE;
                System.out.println("[biTalk收到服务端发来] : " + streamResponse.getResponseInfo());
            }

            //如果连接失败，调用retry重连，消息将有retry.streamObserverImpl重连成功对象处理
            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
                firstConnectedFlag[0] = retry(timer, managedChannel);
                firstConnectedLatch.countDown();
            }

            @Override
            public void onCompleted() {
                System.out.println("onComplated");
            }
        };
        //获取requestStreamObserver
        StreamObserver<StreamRequest> requestStreamObserver = stub.biTalk(streamObserverImpl);

        //客户端发送10次消息
        for (int i = 0; i < 10; i++) {
            requestStreamObserver.onNext(StreamRequest.newBuilder().setRequestInfo(LocalDate.now().toString()).build());
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 重连方法
     *
     * @param timer   单任务定时器
     * @param channel gRPC Channel
     * @return 是否重连成功，返回值仅在第一次连接使用
     */
    boolean retry(ScheduledExecutorService timer, Channel channel) {
        final boolean[] stopRetryTag = {false};
        CountDownLatch latch = new CountDownLatch(RETRY_COUNT);
        for (int i = 0; i < RETRY_COUNT; i++) {
            int finalI = i;
            timer.schedule(() -> {
                System.out.println(Thread.currentThread().getName());
                System.out.printf("第%d次重连中...%n", finalI + 1);
                // 发送重试连接请求
                PersonServiceGrpc.PersonServiceStub stub = PersonServiceGrpc.newStub(channel);
                //定义并实现StreamObserver接口streamObserverImpl为接口实现类
                StreamObserver<StreamResponse> streamObserverImpl = new StreamObserver<StreamResponse>() {
                    @Override
                    public void onNext(StreamResponse streamResponse) {
                        System.out.println("[重连收到服务端发来] : " +streamResponse.getResponseInfo());
                        System.out.println(Thread.currentThread().getName());
                        stopRetryTag[0] = true;
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        System.out.println(throwable.getMessage());
                        if (stopRetryTag[0]){
                            retry(timer, channel);
                        }
                    }

                    @Override
                    public void onCompleted() {
                        System.out.println("onComplated");
                    }
                };
                //获取requestStreamObserver
                StreamObserver<StreamRequest> requestStreamObserver = stub.biTalk(streamObserverImpl);
                requestStreamObserver.onNext(StreamRequest.newBuilder().setRequestInfo(LocalDate.now().toString()+"重连").build());
                latch.countDown();
            }, 0, TimeUnit.SECONDS);

            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            if (stopRetryTag[0]){
                return true;
            }
        }
        try {
            latch.await();
        } catch (InterruptedException ignored) {
        }
        return false;
    }
}